﻿/* Adapted from the Aquiles project: http://aquiles.codeplex.com/. */

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Diagnostics;
using Lucandra.Net;

namespace Lucandra.Net.Connection.Endpoint
{
    internal class RoundRobinEndpointManager : IEndpointManager
    {
        private const int DefaultDueTime = 5000;
        private const int DefaultPeriodicTime = 60000;

        private List<CassandraEndpoint> configuredEndpoints;
        private Queue<CassandraEndpoint> availableEndpoints;
        private HashSet<CassandraEndpoint> bannedEndpoints;
        private Timer endpointRecoveryTimer;

        public RoundRobinEndpointManager()
        {
            this.configuredEndpoints = new List<CassandraEndpoint>();
            this.availableEndpoints = new Queue<CassandraEndpoint>();
            this.bannedEndpoints = new HashSet<CassandraEndpoint>();
            this.endpointRecoveryTimer = new Timer(new TimerCallback(this.EndpointRecoveryTimerCallback), null, Timeout.Infinite, Timeout.Infinite);
        }

        public void AddEndpoint(CassandraEndpoint cassandraEndpoint)
        {
            this.availableEndpoints.Enqueue(cassandraEndpoint);
            this.configuredEndpoints.Add(cassandraEndpoint);
        }

        public IEnumerable<CassandraEndpoint> Endpoints
        {
            get { return this.configuredEndpoints; }
        }

        public CassandraEndpoint Retrieve()
        {
            CassandraEndpoint borrowedEndpoint = null;
            do
            {
                borrowedEndpoint = null;
                lock (this.availableEndpoints)
                {
                    if (this.availableEndpoints.Count > 0)
                        borrowedEndpoint = this.availableEndpoints.Dequeue();
                    else
                        throw new LucandraException("All configured Cassandra endpoints are unavailable.  Cluster may be down.");
                }
            }
            while (borrowedEndpoint != null && this.IsBanned(borrowedEndpoint));

            if (borrowedEndpoint != null)
                lock (this.availableEndpoints)
                    this.availableEndpoints.Enqueue(borrowedEndpoint);

            return borrowedEndpoint;
        }

        public void Ban(CassandraEndpoint endpoint)
        {
            lock (this.bannedEndpoints)
                if (!this.bannedEndpoints.Contains(endpoint))
                    this.bannedEndpoints.Add(endpoint);
        }

        #region Methods [private]

        private bool IsBanned(CassandraEndpoint cassandraEndpoint)
        {
            bool isBanned = false;
            if (this.bannedEndpoints.Count > 0)
            {
                lock (this.bannedEndpoints)
                {
                    isBanned = this.bannedEndpoints.Contains(cassandraEndpoint);
                    this.StartTimer(DefaultDueTime);
                }
            }
            return isBanned;
        }

        private void StopTimer()
        {
            this.endpointRecoveryTimer.Change(Timeout.Infinite, Timeout.Infinite);
        }

        private void StartTimer(int dueTime)
        {
            this.endpointRecoveryTimer.Change(dueTime, Timeout.Infinite);
        }

        #endregion

        #region Timer Callbacks

        private void EndpointRecoveryTimerCallback(object state)
        {
            bool completedGracefully = false;
            int unrecovered = 0;
            int recovered = 0;

            this.StopTimer();

            try
            {
                HashSet<CassandraEndpoint> clonedBanList = null;
                lock (this.bannedEndpoints)
                    clonedBanList = new HashSet<CassandraEndpoint>(this.bannedEndpoints);

                bool isUp = false;
                foreach (var endpoint in clonedBanList)
                {
                    var cassandraClient = CassandraConnectionFactory.Create(endpoint);

                    try
                    {
                        cassandraClient.OpenTransport();
                        cassandraClient.InnerClient.describe_cluster_name();

                        lock (this.bannedEndpoints)
                            this.bannedEndpoints.Remove(endpoint);
                        lock (this.availableEndpoints)
                            this.availableEndpoints.Enqueue(endpoint);

                        recovered++;

                        Debug.WriteLine("Endpoint {0} has come back up and has been moved to the available queue.");
                    }
                    catch
                    {
                        unrecovered++;
                    }
                    finally
                    {
                        cassandraClient.CloseTransport();
                    }
                }

                completedGracefully = true;
            }
            catch (Exception ex)
            {
                Debug.WriteLine(ex);
            }
            finally
            {
                if (completedGracefully)
                {
                    Debug.WriteLine("Endpoint recovery completed successfully.  {0} endpoints recovered, {1} endpoints still down.", recovered, unrecovered);
                    if (unrecovered > 0)
                        this.StartTimer(DefaultPeriodicTime);
                }
                else
                {
                    Debug.WriteLine("Endpoint recovery operation failed due to an exception.  {0} endpoints recovered, {1} endpoints still down.  Operation will be retried in {2} milliseconds.", recovered, unrecovered, DefaultDueTime);
                    this.StartTimer(DefaultDueTime);
                }
            }
        }

        #endregion
    }
}
